package com.zbj.storm.wordcount;

import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Tuple;

import java.io.BufferedWriter;
import java.io.FileWriter;
import java.util.Collections;
import java.util.List;
import java.util.Map;

/**
 * ReportBolt
 *
 * @author weigang
 * @create 2019-09-04
 **/
public class ReportBolt extends BaseRichBolt {

    private OutputCollector collector;

    private Map<String, Long> map;

    @Override
    public void prepare(Map<String, Object> topoConf, TopologyContext context, OutputCollector collector) {
        this.collector = collector;
        map = Maps.newHashMap();
    }

    @Override
    public void execute(Tuple input) {
        try {
            String word = input.getStringByField("word");
            Long count = input.getLongByField("count");
            map.put(word, count);
            print();
            collector.ack(input);
        } catch (Exception e) {
            collector.fail(input);
        }
    }

    private void print() {
        // 动态在test.txt加入内容 不能动态读取
        System.out.println("----------统计结果----------");
        List<String> keys = Lists.newArrayList();
        keys.addAll(map.keySet());
        Collections.sort(keys);

        try {
            BufferedWriter writer = new BufferedWriter(new FileWriter(this.getClass().getSimpleName() + ".txt"));
            for (String key : keys) {
                System.out.println(key + "-> " + map.get(key));
                writer.write(key + "-> " + map.get(key) + "\n");
            }
            writer.flush();
            writer.close();
        } catch (Exception e) {

        }
        System.out.println("--------------------");
    }

    //
    ///**
    // * 当IBolt即将关闭时调用。 如果工人关闭有序，Storm会尽力调用此方法
    // */
    //@Override
    //public void cleanup() {
    //    System.out.println("----------统计结果----------");
    //    List<String> keys = Lists.newArrayList();
    //    keys.addAll(map.keySet());
    //    Collections.sort(keys);
    //    for (String key : keys) {
    //        System.out.println(key + "-> " + map.get(key));
    //    }
    //    System.out.println("--------------------");
    //}

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        //declarer.declare(new Fields("word", "count"));
    }
}